package cn.ac.iie.di.ban.ycl.cleaner.sender;

import cn.ac.iie.di.ban.ycl.cleaner.CleanerMain;
import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.apache.commons.configuration2.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn> <zhangmingcheng@iie.ac.cn>
 * @date 2016-11-8 11:14:11
 * @version V1.0
 * @desc
 */
public class DeduplicationMessageSenderThread implements Runnable {

	private static final Logger logger;
	private static final Logger backupLog;

	static {
		logger = LoggerFactory.getLogger(DeduplicationMessageSenderThread.class);
		backupLog = LoggerFactory.getLogger("backup");
	}
	MessageSender sender;
	long timeOut;
	long batchNum;
	List<Object[]> cacheDatas = new ArrayList<>();

	public DeduplicationMessageSenderThread(Configuration conf) {
		String nameServer = conf.getString(Constants.RECEIVER_MQ_NAMESERVER);
		String topic = conf.getString(Constants.DEDUPLICATION_SEND_TOPIC);
		sender = new MessageSender(nameServer, topic);
		timeOut = conf.getLong(Constants.SEND_QUEUE_TIME_OUT, 5000);
		batchNum = conf.getLong(Constants.SEND_QUEUE_BATCH_NUM, 50);
	}

	@Override
	public void run() {
		logger.info("deduplication send thread start!");
		long startTime = System.currentTimeMillis();
		loop:
		while (true) {
			try {
				REMessage message = (REMessage) CleanerMain.deduplicationQueue.take();
				if (message == null) {
					logger.warn("deduplication message is null!");
					continue;
				}
				List<Object[]> datas = message.data;
				if (datas == null || datas.isEmpty()) {
					logger.warn("deduplication message datas is null!");
					continue;
				}
//                cacheDatas.addAll(datas);
//                long endTime = System.currentTimeMillis();
//                long time = endTime - startTime;
//                if (time < timeOut && cacheDatas.size() < batchNum) {
//                    continue;
//                }
//                REMessage cacheMessage = new REMessage(message.colNames,
//                        message.types, cacheDatas, message.docSchemaName,
//                        message.userDesc);
//                sender.sendMessage(cacheMessage);
				for (int i = 0; i < 5; ++i) {
					try {
						sender.sendMessage(message);
						logger.info("send deduplication messages num: {}", message.data.size());
						continue loop;
					} catch (Exception e) {
						logger.warn("send error for the " + i + "th trial.", e);
					}
				}
				logger.warn("send error after 5 trials, log it to backup file.");
				while (true) {
					try {
						printColumnAndData(message);
						printColumnAndData(message);
						break;
					} catch (Exception e) {
						logger.warn("log data to file error .");
					}
				}
//				cacheDatas.clear();
				startTime = System.currentTimeMillis();
			} catch (Exception ex) {
				logger.error("send deduplication messages error!", ex);
			}
		}
	}

	private static String getJsong(REMessage msg) {
		JSONObject obj = new JSONObject();
		JSONObject type = new JSONObject();
		for (int i = 0; i < msg.colNames.length; ++i) {
			type.put(msg.colNames[i], msg.types[i].name());
		}
		obj.put("type", type);
		JSONArray data = new JSONArray();
		for (Object[] rec : msg.data) {
			JSONObject jsonRec = new JSONObject();
			for (int i = 0; i < msg.colNames.length; ++i) {
				switch (msg.types[i]) {
					case Boolean:
					case Int:
					case Long:
					case Float:
					case Double:
					case String:
						jsonRec.put(msg.colNames[i], rec[i]);
						break;

					case Struct:
						JSONObject map = new JSONObject((Map) rec[i]);
						jsonRec.put(msg.colNames[i], map);
						break;
					case Structs:
						JSONArray v = new JSONArray();
						for (Map m : (Map[]) rec[i]) {
							v.add(new JSONObject(m));
						}
						jsonRec.put(msg.colNames[i], v);
						break;
					case Doc:
					case Docs:
						break;

					case Booleans:
					case Binary:
					case Doubles:
					case Floats:
					case Ints:
					case Longs:
					case Strings:
						List<Object> list = new ArrayList<>();
						int n = Array.getLength(rec[i]);
						for (int j = 0; j < n; ++j) {
							list.add(Array.get(rec[i], j));
						}
						JSONArray a = new JSONArray(list);
						jsonRec.put(msg.colNames[i], a);
						break;
					case Binaries:
						JSONArray bss = new JSONArray();
						for (byte[] bs : (byte[][]) rec[i]) {
							list = new ArrayList<>();
							for (byte b : bs) {
								list.add(b);
							}
							bss.add(new JSONArray(list));
						}
						jsonRec.put(msg.colNames[i], bss);
						break;
				}
			}
			data.add(rec);
		}
		obj.put("data", data);
		return obj.toJSONString();
	}

	private void printColumnAndType(REMessage message) {
		Object[] columns = message.colNames;
		Object[] types = message.types;
		StringBuilder sb = new StringBuilder();
		sb.append(" {");
		for (int i = 0; i < columns.length; i++) {
			if (i > 0) {
				sb.append(",");
			}
			sb.append(columns[i] + ":" + types[i]);
		}
		sb.append("}");
		backupLog.info(sb.toString());
	}

	private void printColumnAndData(REMessage message) {
		Object[] columns = message.colNames;
		Object[] types = message.types;
		for (Object[] data : message.data) {
			printSinaleColumnAndData(columns, types, data);
		}
	}

	private void printSinaleColumnAndData(Object[] columns, Object[] types, Object[] data) {
		StringBuilder sb = new StringBuilder();
		sb.append(" {");
		for (int i = 0; i < columns.length; i++) {
			if (i > 0) {
				sb.append(",");
			}
			if (types[i].equals(REFieldType.Strings)) {
				sb.append("\"" + columns[i] + "\":\"" + Arrays.toString((String[]) data[i]) + "\"");
			} else if (types[i].equals(REFieldType.Structs)) {
				sb.append("\"" + columns[i] + "\":\"" + Arrays.toString((Map[]) data[i]) + "\"");
			} else if (types[i].equals(REFieldType.Ints) || types[i].equals(REFieldType.Longs)) {
				sb.append("\"" + columns[i] + "\":\"" + Arrays.toString((Object[]) data[i]) + "\"");
			} else {
				sb.append("\"" + columns[i] + "\":\"" + data[i] + "\"");
			}
		}
		sb.append("}");
		backupLog.info(sb.toString());
	}
}
